package top.zhaoyuliu.logbook.server.connect.client;

import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.Semaphore;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock.ReadLock;
import java.util.stream.Collectors;

import org.springframework.data.mongodb.core.MongoTemplate;
import org.tio.client.ClientChannelContext;
import org.tio.client.ClientGroupContext;
import org.tio.client.TioClient;
import org.tio.core.ChannelContext;
import org.tio.core.Tio;
import org.tio.utils.lock.SetWithLock;

import cn.hutool.core.collection.CollectionUtil;
import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;
import top.zhaoyuliu.logbook.client.builder.LbBuilder;
import top.zhaoyuliu.logbook.client.connect.packet.LbPacket;
import top.zhaoyuliu.logbook.client.connect.packet.PacketBuilder;
import top.zhaoyuliu.logbook.client.connect.packet.PacketType;
import top.zhaoyuliu.logbook.client.constant.Const;
import top.zhaoyuliu.logbook.client.utils.IPUtils;
import top.zhaoyuliu.logbook.server.connect.builder.ElectionBuilder;
import top.zhaoyuliu.logbook.server.connect.builder.Node;
import top.zhaoyuliu.logbook.server.connect.thread.HandlerThread;
import top.zhaoyuliu.logbook.server.connect.thread.HandlerThreadPool;
import top.zhaoyuliu.logbook.server.connect.thread.TimeThreadPool;
import top.zhaoyuliu.logbook.server.properties.LbServiceProperties;

/**
 * Copyright © 2021 Arain. All rights reserved.
 *
 * @Title: ClientStart.java
 * @Prject: logbook-server
 * @Package: top.zhaoyuliu.logbook.server.connect.client
 * @Description: <功能详细描述>
 * @author: Arain
 * @date: 2021年1月24日 上午11:58:54
 * @version: V1.0
 */
@Slf4j
@Getter
@Setter
public class ClientStart {

	private MongoTemplate mongoTemplate;

	private ClientGroupContext clientGroupContext;

	private LbServiceProperties lbServiceProperties;

	private volatile Map<String, Integer> nodeTask = new ConcurrentHashMap<>();

	private volatile AtomicInteger taskNum = new AtomicInteger(0);;

	private volatile BlockingQueue<LbBuilder> logQueue = new LinkedBlockingQueue<>();

	private volatile BlockingQueue<LbBuilder> paraQueue = new LinkedBlockingQueue<>();

	private volatile BlockingQueue<LbBuilder> trackQueue = new LinkedBlockingQueue<>();

	private volatile Map<String, ChannelContext> nodeChannelContext = new ConcurrentHashMap<>();

	private Semaphore logQueuesemaphore = new Semaphore(0);
	private Semaphore trackQueuesemaphore = new Semaphore(0);
	private Semaphore paraQueueQueuesemaphore = new Semaphore(0);

	public ClientStart(ClientGroupContext clientGroupContext, LbServiceProperties lbServiceProperties) {
		this.clientGroupContext = clientGroupContext;
		this.lbServiceProperties = lbServiceProperties;
	}

	public void start() {
		List<String> cluster = lbServiceProperties.getCluster();
		if (CollectionUtil.isNotEmpty(cluster)) {
			Set<Node> serverNodes = new HashSet<>();
			for (String address : cluster) {
				String[] add = address.split(":");
				if (add.length != 2) {
					continue;
				}
				try {
					String ip = add[0];
					Integer port = Integer.parseInt(add[1]);
					if (IPUtils.getIpAdd().equals(add[0])
							&& lbServiceProperties.getPort().intValue() == port.intValue()) {
						continue;
					}
					Node node = new Node(ip, port);
					serverNodes.add(node);
				} catch (Exception e) {
					log.warn("连接失败 -> {}", address);
					continue;
				}
			}
			bindServerGroup(serverNodes);
		}
		handlerData();
	}

	private void bindServerGroup(Set<Node> serverNodes) {
		SetWithLock<ChannelContext> setWithLock = Tio.getAllChannelContexts(clientGroupContext);
		Lock lock2 = setWithLock.getLock().readLock();
		lock2.lock();
		try {
			Set<ChannelContext> set = setWithLock.getObj();
			Set<org.tio.core.Node> connectedNodes = set.stream().map(ChannelContext::getServerNode)
					.collect(Collectors.toSet());
			for (Node node : serverNodes) {
				if (!connectedNodes.contains(node)) {
					try {
						TioClient client = new TioClient(clientGroupContext);
						startHeartbeatTask();
						ClientChannelContext channelContext = client.connect(node);
						channelContext.setBindIp(node.getIp());
						channelContext.setBindPort(node.getPort());
					} catch (Exception e) {
						e.printStackTrace();
					}
				}
			}
			for (ChannelContext channelContext : set) {
				org.tio.core.Node node = channelContext.getServerNode();
				if (!serverNodes.contains(node)) {
					Tio.remove(channelContext, "主动关闭" + node.getIp());
				}
			}
			Const.CLUSTER = true;
		} finally {
			lock2.unlock();
		}
	}

	public void startHeartbeatTask() {
		TimeThreadPool.INSTANCE.executor().scheduleWithFixedDelay(new Runnable() {
			@Override
			public void run() {
				if (!clientGroupContext.isStopped()) {
					SetWithLock<ChannelContext> setWithLock = clientGroupContext.connecteds;
					ReadLock readLock = setWithLock.readLock();
					readLock.lock();
					try {
						Set<ChannelContext> set = setWithLock.getObj();
						for (ChannelContext entry : set) {
							ClientChannelContext channelContext = (ClientChannelContext) entry;
							String node = channelContext.getServerNode().getIp() + ":"
									+ channelContext.getServerNode().getPort();
							if (channelContext.isClosed || channelContext.isRemoved) {
								try {
									nodeTask.remove(node);
									nodeChannelContext.remove(node);
								} catch (Throwable e) {
									log.error("", e);
								}
								continue;
							}
							nodeChannelContext.put(node, channelContext);
							log.info("当前任务数 -> {} ,发送心跳包 -> {}", taskNum.get(), channelContext.toString());
							String me = IPUtils.getIpAdd() + ":" + lbServiceProperties.getPort();
							ElectionBuilder electionBuilder = ElectionBuilder.builder().taskNum(taskNum.get()).addr(me)
									.build();
							LbPacket build = new PacketBuilder().setType(PacketType.HEART_BEAT).setBody(electionBuilder)
									.build();
							Tio.send(channelContext, build);
						}
					} catch (Throwable e) {
						log.error("", e);
					} finally {
						try {
							readLock.unlock();
						} catch (Throwable e) {
							log.error(e.toString(), e);
						} finally {

						}
					}
				}
			}
		}, 3000, 1000, TimeUnit.MILLISECONDS);
	}

	private void handlerData() {
		HandlerThreadPool.INSTANCE.executor().execute(new HandlerThread(logQueue, logQueuesemaphore, this));
		HandlerThreadPool.INSTANCE.executor().execute(new HandlerThread(trackQueue, trackQueuesemaphore, this));
		HandlerThreadPool.INSTANCE.executor().execute(new HandlerThread(paraQueue, paraQueueQueuesemaphore, this));
	}
}
